package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo24MapJoin {
  def main(args: Array[String]): Unit = {
    // score分数数据（大） 关联 students学生数据（小）
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo24MapJoin")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    // 基于students数据构建Map
    val stuRDD: RDD[String] = sc.textFile("Spark/data/students.txt")
    val stuMap: Map[String, String] = stuRDD
      .map(line => (line.split(",")(0), line))
      .collect()
      .toMap

    // 使用SparkContext进行广播
    val broStuMap: Broadcast[Map[String, String]] = sc.broadcast(stuMap)

    // 遍历Score数据
    sc.textFile("Spark/data/score.txt")
      .map(line => {
        val id: String = line.split(",")(0)

        // 使用广播变量实现MapJoin
        val stu: String = broStuMap.value.getOrElse(id, "")
        (line, stu)
      })
      .foreach(println)


  }

}
